package com.mjf.app.dwm;

import cn.hutool.core.date.DateUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mjf.app.function.DimAsyncFunction;
import com.mjf.bean.OrderDetail;
import com.mjf.bean.OrderInfo;
import com.mjf.bean.OrderWide;
import com.mjf.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;
import java.util.concurrent.TimeUnit;

/**
 * 使用订单信息表与订单详情表(双流join)生成订单宽表，并使用维度表(使用redis做热点缓存，异步IO)补充订单宽表信息
 *
 * 数据流向：web/app -> nginx -> springboot -> MySQL -> flinkApp -> kafka(ods) -> flinkApp -> kafka(dwd)/HBase(dim) -> flinkApp(redis) -> Kafka(dwm)
 * 程序：mockDb(包含web/nginx/springboot) -> MySQL -> FlinkCDC(包含flinkApp/kafka(ods)) -> BaseDbApp(包含flinkApp/kafka(dwd)/HBase(dim)) -> OrderWideApp(包含flinkApp/Kafka(dwm))
 * 环境：hdfs zookeeper kafka hbase phoenix(bin/sqlline.py) redis
 */
public class OrderWideApp {
    public static void main(String[] args) throws Exception {

        // 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

/*
        // 测试时关闭
        // 设置状态后端
        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9000/gmall-flink/checkpoint"));
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(10000L);
        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);

        // 设置重启策略
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 10));
*/

        // 2.读取 Kafka dwd_order_info/dwd_order_detail 主题数据，并转换为 JavaBean 对象&提取时间戳生成 watermark
        String groupId = "OrderWideApp1";
        String orderInfoTopic = "dwd_order_info";
        String orderDetailTopic = "dwd_order_detail";
        String sinkTopic = "dwm_order_wide";

        SingleOutputStreamOperator<OrderInfo> orderInfoDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderInfoTopic, groupId))
                .map(line -> {
                    OrderInfo orderInfo = JSON.parseObject(line, OrderInfo.class);
                    String create_time = orderInfo.getCreate_time();
                    String[] dateArr = create_time.split(" ");
                    orderInfo.setCreate_date(dateArr[0]);
                    orderInfo.setCreate_hour(dateArr[1].split(":")[0]);
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    orderInfo.setCreate_ts(sdf.parse(create_time).getTime());
                    return orderInfo;
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderInfo>forMonotonousTimestamps().withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderInfo>() {
                                    @Override
                                    public long extractTimestamp(OrderInfo element, long recordTimestamp) {
                                        return element.getCreate_ts();
                                    }
                                }
                        )
                );

        SingleOutputStreamOperator<OrderDetail> orderDetailDS = env.addSource(MyKafkaUtil.getKafkaConsumer(orderDetailTopic, groupId))
                .map(line -> {
                    OrderDetail orderDetail = JSON.parseObject(line, OrderDetail.class);
                    String create_time = orderDetail.getCreate_time();
                    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                    orderDetail.setCreate_ts(sdf.parse(create_time).getTime());
                    return orderDetail;
                }).assignTimestampsAndWatermarks(
                        WatermarkStrategy.<OrderDetail>forMonotonousTimestamps().withTimestampAssigner(
                                new SerializableTimestampAssigner<OrderDetail>() {
                                    @Override
                                    public long extractTimestamp(OrderDetail element, long recordTimestamp) {
                                        return element.getCreate_ts();
                                    }
                                }
                        )
                );

        // 3.双流 join
        SingleOutputStreamOperator<OrderWide> orderWideWithNoDimDS = orderInfoDS.keyBy(OrderInfo::getId)
                .intervalJoin(orderDetailDS.keyBy(OrderDetail::getOrder_id))
                .between(Time.seconds(-5), Time.seconds(5)) // 生产环境中给的时间为最大延迟时间
                .process(new ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>() {
                    @Override
                    public void processElement(OrderInfo orderInfo, OrderDetail orderDetail, ProcessJoinFunction<OrderInfo, OrderDetail, OrderWide>.Context ctx, Collector<OrderWide> out) throws Exception {
                        out.collect(new OrderWide(orderInfo, orderDetail));
                    }
                });

//        orderWideWithNoDimDS.print("orderWideWithNoDimDS===>"); // 测试双流Join效果

        // 4.关联维度信息
        // map方法为同步IO，需要优化为异步IO，减少与外部系统的交互时间
/*        orderWideWithNoDimDS.map(orderWide ->{
            // 关联用户维度
            Long user_id = orderWide.getUser_id();
            // 根据 user_id 到 phoenix 中查询用户信息，补充 orderWide 中的维度数据
            // 地区信息
            // SKU
            // SPU
            // ...
            return orderWide
        });*/

        // 4.1关联用户维度
        SingleOutputStreamOperator<OrderWide> orderWideWithUser = AsyncDataStream.unorderedWait(
                orderWideWithNoDimDS,
                new DimAsyncFunction<OrderWide>("DIM_USER_INFO") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getUser_id().toString();
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfo) {
                        String birthday = dimInfo.getString("BIRTHDAY");
                        String gender = dimInfo.getString("GENDER");
                        int age = DateUtil.ageOfNow(birthday);  // 生日转年龄
                        orderWide.setUser_age(age);
                        orderWide.setUser_gender(gender);
                    }
                },
                1,  // 访问zk的超时时间为60s,这里设置的超时时间要大于60s
                TimeUnit.MINUTES
        );

//        orderWideWithUser.print("orderWideWithUser===>");  // 测试关联用户维度后结果

        // 4.2关联地区维度
        SingleOutputStreamOperator<OrderWide> orderWideWithProvince = AsyncDataStream.unorderedWait(
                orderWideWithUser,
                new DimAsyncFunction<OrderWide>("DIM_BASE_PROVINCE") {
                    @Override
                    public String getKey(OrderWide orderWide) {
                        return orderWide.getProvince_id().toString();
                    }

                    @Override
                    public void join(OrderWide orderWide, JSONObject dimInfo) {
                        orderWide.setProvince_name(dimInfo.getString("NAME"));
                        orderWide.setProvince_area_code(dimInfo.getString("AREA_CODE"));
                        orderWide.setProvince_iso_code(dimInfo.getString("ISO_CODE"));
                        orderWide.setProvince_3166_2_code(dimInfo.getString("ISO_3166_2"));
                    }
                },
                1L,
                TimeUnit.MINUTES
        );

        // 4.3关联SKU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSku =
                AsyncDataStream.unorderedWait(
                        orderWideWithProvince,
                        new DimAsyncFunction<OrderWide>("DIM_SKU_INFO") {
                            @Override
                            public void join(OrderWide orderWide, JSONObject jsonObject) {
                                orderWide.setSku_name(jsonObject.getString("SKU_NAME"));
                                orderWide.setCategory3_id(jsonObject.getLong("CATEGORY3_ID"));
                                orderWide.setSpu_id(jsonObject.getLong("SPU_ID"));
                                orderWide.setTm_id(jsonObject.getLong("TM_ID"));
                            }

                            @Override
                            public String getKey(OrderWide orderWide) {
                                return String.valueOf(orderWide.getSku_id());
                            }
                        },
                        1L,
                        TimeUnit.MINUTES);


        // 4.4关联SPU维度
        SingleOutputStreamOperator<OrderWide> orderWideWithSpu =
                AsyncDataStream.unorderedWait(
                        orderWideWithSku,
                        new DimAsyncFunction<OrderWide>("DIM_SPU_INFO") {
                            @Override
                            public void join(OrderWide orderWide, JSONObject jsonObject) {
                                orderWide.setSpu_name(jsonObject.getString("SPU_NAME"));
                            }

                            @Override
                            public String getKey(OrderWide orderWide) {
                                return String.valueOf(orderWide.getSpu_id());
                            }
                        },
                        1L,
                        TimeUnit.MINUTES);


        // 4.5关联TradeMark维度
        SingleOutputStreamOperator<OrderWide> orderWideWithTm =
                AsyncDataStream.unorderedWait(
                        orderWideWithSpu,
                        new DimAsyncFunction<OrderWide>("DIM_BASE_TRADEMARK") {
                            @Override
                            public void join(OrderWide orderWide, JSONObject jsonObject) {
                                orderWide.setTm_name(jsonObject.getString("TM_NAME"));
                            }

                            @Override
                            public String getKey(OrderWide orderWide) {
                                return String.valueOf(orderWide.getTm_id());
                            }
                        },
                        1L,
                        TimeUnit.MINUTES);

        // 4.6关联Category维度
        SingleOutputStreamOperator<OrderWide> orderWideWithCategory3 =
                AsyncDataStream.unorderedWait(
                        orderWideWithTm,
                        new DimAsyncFunction<OrderWide>("DIM_BASE_CATEGORY3") {
                            @Override
                            public void join(OrderWide orderWide, JSONObject jsonObject) {
                                orderWide.setCategory3_name(jsonObject.getString("NAME"));
                            }

                            @Override
                            public String getKey(OrderWide orderWide) {
                                return String.valueOf(orderWide.getCategory3_id());
                            }
                        },
                        1L,
                        TimeUnit.MINUTES);

        orderWideWithCategory3.print("orderWideWithCategory3===>"); // 测试维度表关联结束后的结果

        // 5.将数据写入 Kafka
        orderWideWithCategory3.map(JSONObject::toJSONString).addSink(MyKafkaUtil.getKafkaProducer(sinkTopic));

        // 6.启动任务
        env.execute(OrderWideApp.class.getName());

    }
}
